package com.sleep.cloudstreamdemo.consumer;

import cn.hutool.core.lang.Assert;
import com.alibaba.cloud.stream.binder.rocketmq.integration.inbound.pull.RocketMQAckCallback;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.sleep.cloudstreamdemo.domian.DemoMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.acks.AcknowledgmentCallback;
import org.springframework.integration.acks.SimpleAcknowledgment;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.function.Consumer;

/**
 * @ClassName DemoConsumer
 * @Description TODO
 * @Author 郭敖
 * @Date 2024年04月18日 9:29
 * @Version 1.0
 */
@Component
@Slf4j
public class DemoConsumer {
    @Bean
    Consumer<Message<DemoMessage>> demo(){
        return message ->{
            log.info("接收到消息：{}",message.getPayload());
            log.info("消息的头部tag：{}",message.getHeaders().get(AmqpHeaders.DELIVERY_TAG));
            Channel channel = message.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
            try {
                assert channel != null;
                // todo: 通过 channel.basicAck() 签收消息 手动确认
                channel.basicAck(message.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class),false);
            } catch (IOException e) {
                try {
                    // 消息重新进入队列
                    channel.basicNack(message.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class),false,true);
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
                throw new RuntimeException(e);
            }

        };
    }

    /**
     * 通用的消息手动确认
     * @return
     */
    @Bean
    Consumer<Message<DemoMessage>> demo2(){
        return message ->{
            // 通过StaticMessageHeaderAccessor获取和操作消息的请求头
            AcknowledgmentCallback callback = StaticMessageHeaderAccessor.getAcknowledgmentCallback(message);
            assert callback != null;
            try{
                // 获取消息体
                DemoMessage payload = message.getPayload();
                // 消费者处理消息
                // 手动确认消息
                callback.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
            }catch (Exception e){
                callback.acknowledge(AcknowledgmentCallback.Status.REJECT);
            }
        };
    }
}
